#ifndef HASHACTIVITYBOX_H_JM_20120910
#define HASHACTIVITYBOX_H_JM_20120910

#include "../RootBox.h"
#include "HashBox.h"
#include "../../common/hash/BoostHashWrapper.h"
#include "../../common/Utils.h"

#include <bobox_box.hpp>
#include <bobox_types.hpp>

#include <sstream>


namespace jm4bbx {

	struct HashActivityBoxDescriptor {
		static const  groupType INPUT_GROUP;
		static const  groupType OUTPUT_GROUP;	
		static const  columnType INPUT_COLUMN;
		static const  columnType OUTPUT_COLUMN;
		static const  inType IN_ARC;
		static const  outType OUT_ARC;

		static const char* RESULT_CHUNK_SIZE_PARAMETER_NAME;
	};

	template</*class BoxDescriptor,*/ class hashBoxType>
	class  HashActivityBox : public Box/*<BoxDescriptor>*/ {

	public:
		typedef HashActivityBoxDescriptor BoxIO;

		typedef typename hashBoxType::StaticHashTableHolderType StaticHashTableHolderType;
		typedef typename StaticHashTableHolderType::HashTableType::HashEntry HashEntry;
		typedef typename StaticHashTableHolderType::HashTableType::ExtendedHashEntry ExtendedHashEntry;

		typedef typename hashBoxType::policyType policyType;
		typedef typename hashBoxType::valueType valueType;

		HashActivityBox(bobox::box_id_type id, bobox::box_tid_type tid, bobox::request *req, 
			const bobox::parameters &params) 
			: Box(id, tid, req, params)
			, resultChunkSize_(10) {

				if (params.has_parameter(
					BoxIO::RESULT_CHUNK_SIZE_PARAMETER_NAME)) {
						params.get_parameter(
							BoxIO::RESULT_CHUNK_SIZE_PARAMETER_NAME, 
							resultChunkSize_);
				}
		}

		virtual void initImpl() {

			add_group(BoxIO::INPUT_GROUP, 
				BoxIO::INPUT_COLUMN, 
				get_input_descriptors(BoxIO::IN_ARC));

			add_group(BoxIO::OUTPUT_GROUP, 
				BoxIO::OUTPUT_COLUMN, 
				get_output_descriptors(BoxIO::OUT_ARC));

			allocateResultChunk();

			wait4Receive();
		}

		virtual void sync_mach_etwas() {

			Group ig = get_group(BoxIO::INPUT_GROUP);
			if (ig.isPoisoned()/*poisoned_*/) {
				send_poisoned(BoxIO::OUT_ARC);
				return;
			}

			const int rpos = ig.getReadPosition();//rdpos_;
			const int rend = ig.getReadEnd();//rdend_;

			BoostHashWrapper<valueType> hash;


			tableHolder_.getHashTable()->setSynchronized(synchronize());

			auto bid = getBoxInstanceId();

			int counter = 0;
			//for (; ig.rdpos_ < ig.rdend_ && counter < resultChunkSize_; 
			//	++counter, ++ig.rdpos_) {
			for (; ig.getReadPosition() < ig.getReadEnd() && counter < resultChunkSize_; 
				++counter, ig.moveReadPosition()) {
					auto key = get_data<policyType>(BoxIO::INPUT_COLUMN)[ig.getReadPosition()/*rdpos_*/];

					std::stringstream stream;
					stream << key << "@" << ig.getReadPosition();//rdpos_;

					tableHolder_.getHashTable()->foreachExtended(key, [&](ExtendedHashEntry& entry) -> bool {
						handleRecord(key, entry);
						return true;
					}, 0);			

				get_data<bobox::string_policy>(BoxIO::OUTPUT_COLUMN)[counter] = stream.str();
			}

			Group og = get_group(BoxIO::OUTPUT_GROUP);

			og.setWriteBegin(0);//wrbegin_ = 0;
			og.setWritePosition(counter);//wrpos_ = counter;
			og.setWriteEnd(counter);//wrend_ = counter;

			send(BoxIO::OUT_ARC, BoxIO::OUTPUT_GROUP);

			//if (!(ig.rdpos_ < ig.rdend_)) {
			if (!(ig.getReadPosition() < ig.getReadEnd())) {
				sendSignalsOut();
				send_poisoned(BoxIO::OUT_ARC);
				wait4Receive();
			}

		}


	//void add


protected:
	virtual void handleRecord(valueType value, ExtendedHashEntry& entry) = 0; 
	virtual bool synchronize() const = 0;
private:

	StaticHashTableHolderType tableHolder_;
	int resultChunkSize_;

	void wait4Receive() {
		receive(BoxIO::IN_ARC, BoxIO::INPUT_GROUP);
	}

	void allocateResultChunk() {
		allocate(BoxIO::OUTPUT_GROUP, resultChunkSize_);
		get_group(BoxIO::OUTPUT_GROUP).setWritePosition(resultChunkSize_);//wrpos_ = resultChunkSize_;
	}
};

} // namespace jm4bbx


#endif
